package com.youzhi.message;

import com.youzhi.message.dto.Param;
import com.youzhi.message.kafka.KafkaMessageConsumers;
import lombok.extern.slf4j.Slf4j;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

@Slf4j
public class WebSocketServer {

    private final int port;

    public WebSocketServer(int port) {
        this.port = port;
    }

    public void run() {
        // Configure the server.
        ServerBootstrap bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));

        // Set up the event pipeline factory.
        bootstrap.setPipelineFactory(new WebSocketServerPipelineFactory());

        // Bind and start to accept incoming connections.
        bootstrap.bind(new InetSocketAddress(port));

        System.out.println("Web socket server started at port " + port + '.');
        System.out.println("Open your browser and navigate to http://localhost:" + port + '/');
    }


    public static String PORT = "port";

    public static String DEFAULT_PORT = "8085";

    public static String BOOTSTRAP_SERVERS = "bootstrap.servers";

    public static String DEFAULT_BOOTSTRAP_SERVERS = "127.0.0.1:9092";


    public static void main(String[] args) {


        Map<String, String> parseArgs = parseArgs(args);

        String port = parseArgs.getOrDefault(Param.PORT.key, Param.PORT.defaultValue);

        String kafkaServer = parseArgs.getOrDefault(Param.BOOTSTRAP_SERVERS.key, Param.BOOTSTRAP_SERVERS.defaultValue);

        new KafkaMessageConsumers(kafkaServer);

        new WebSocketServer(Integer.parseInt(port)).run();

        log.info("message服务启动成功 port={},bootstrap.servers={}",port,kafkaServer);

    }


    public static Map<String, String> parseArgs(String[] args) {
        Map<String, String> map = new HashMap<>();
        if (args.length > 0) {
            for (int i = 0; i < args.length; i++) {
                String arg = args[i];
                String[] split = arg.split("=");
                map.put(split[0].substring(2), split[1]);
            }
        }
        return map;
    }

}
